package com.demo.kafka.config;

import com.demo.kafka.constants.DelayTopicEnum;
import com.demo.kafka.entity.KafkaMessage;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.support.serializer.JsonDeserializer;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * Kafka客户端配置
 * @author cyj
 */
@Configuration
public class KafkaConfig {

    /**
     * 延迟消息消费端配置
     * @return 消费端
     */
    @Bean("delayConsumer")
    public Consumer<String, KafkaMessage> createConsumer(KafkaProperties properties) {
        KafkaProperties.Consumer consumer = properties.getConsumer();
        String servers = String.join(",", properties.getBootstrapServers());

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, consumer.getGroupId());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 配置指定位移消费
        // earliest ：当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，从头开始消费。
        // latest ：当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，消费该分区下新产生的数据。（从consumer创建开始，后生产的数据）
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        // 指定consumer两次poll的最大时间间隔
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "5000");

        JsonDeserializer<KafkaMessage> jsonDeserializer = new JsonDeserializer<>();
        // 设置消费端可信任packages, '*'是所有包都信任
        jsonDeserializer.trustedPackages("*");

        return new KafkaConsumer<>(props, new StringDeserializer(), jsonDeserializer);
    }

    /**
     * 用于consumer assign分区
     * @param properties kafka配置
     * @return consumer对象
     */
    @Bean("assignConsumer")
    public Consumer<String, KafkaMessage> createAssignConsumer(KafkaProperties properties) {
        String servers = String.join(",", properties.getBootstrapServers());

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "assignConsumer");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 配置指定位移消费
        // earliest ：当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，从头开始消费。
        // latest ：当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，消费该分区下新产生的数据。（从consumer创建开始，后生产的数据）
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        // 指定consumer两次poll的最大时间间隔
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "5000");

        JsonDeserializer<KafkaMessage> jsonDeserializer = new JsonDeserializer<>();
        // 设置消费端可信任packages, '*'是所有包都信任
        jsonDeserializer.trustedPackages("*");

        return new KafkaConsumer<>(props, new StringDeserializer(), jsonDeserializer);
    }

    /**
     * 延迟消息消费端配置
     * @return 消费端
     */
    @Bean("delayConsumerMap")
    public Map<String, Consumer<String, KafkaMessage>> createDelayConsumer(KafkaProperties properties) {

        DelayTopicEnum[] values = DelayTopicEnum.values();
        int length = values.length;
        Map<String, Consumer<String, KafkaMessage>> map = new HashMap<>(length);

        for (int i = 0; i < length; i++) {
            DelayTopicEnum topicEnum = values[i];

            KafkaProperties.Consumer consumer = properties.getConsumer();
            String servers = String.join(",", properties.getBootstrapServers());

            Properties props = new Properties();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
            props.put(ConsumerConfig.GROUP_ID_CONFIG, consumer.getGroupId() + "_" + i);
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
            // 配置指定位移消费
            // earliest ：当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，从头开始消费。
            // latest ：当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，消费该分区下新产生的数据。（从consumer创建开始，后生产的数据）
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
            props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
            // 指定consumer两次poll的最大时间间隔
            props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "5000");

            JsonDeserializer<KafkaMessage> jsonDeserializer = new JsonDeserializer<>();
            // 设置消费端可信任packages, '*'是所有包都信任
            jsonDeserializer.trustedPackages("*");

            KafkaConsumer<String, KafkaMessage> kafkaConsumer = new KafkaConsumer<>(props, new StringDeserializer(), jsonDeserializer);
            kafkaConsumer.subscribe(List.of(topicEnum.getTopic()));

            map.put(topicEnum.getTopic(), kafkaConsumer);
        }

        return map;
    }
}
